package com.electric.flinkcep.service;

import com.electric.flinkcep.model.LoadPredictionEvent;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.util.List;
import java.util.Map;

/**
 * 电力负荷预测CEP服务
 * 实现连续5次误差较大的告警逻辑
 */
public class LoadPredictionCEPService {
    
    /**
     * 应用电力负荷预测误差监控模式
     * 当连续5次预测误差较大时输出告警
     */
    public static DataStream<String> applyLoadPredictionPattern(DataStream<LoadPredictionEvent> eventStream) {
        
        // 定义CEP模式：连续5次误差较大
        Pattern<LoadPredictionEvent, ?> largeErrorPattern = Pattern
                .<LoadPredictionEvent>begin("first")
                .where(new SimpleCondition<LoadPredictionEvent>() {
                    @Override
                    public boolean filter(LoadPredictionEvent event) {
                        boolean isLarge = event.isLargeError();
                        System.out.println("CEP检查事件: 时间=" + event.getTrueData().getTimestamp() + 
                                         ", 误差=" + event.getRelativeErrorPercentage() + "%, 是否大误差=" + isLarge);
                        return isLarge;
                    }
                })
                .next("second")
                .where(new SimpleCondition<LoadPredictionEvent>() {
                    @Override
                    public boolean filter(LoadPredictionEvent event) {
                        return event.isLargeError();
                    }
                })
                .next("third")
                .where(new SimpleCondition<LoadPredictionEvent>() {
                    @Override
                    public boolean filter(LoadPredictionEvent event) {
                        return event.isLargeError();
                    }
                })
                .next("fourth")
                .where(new SimpleCondition<LoadPredictionEvent>() {
                    @Override
                    public boolean filter(LoadPredictionEvent event) {
                        return event.isLargeError();
                    }
                })
                .next("fifth")
                .where(new SimpleCondition<LoadPredictionEvent>() {
                    @Override
                    public boolean filter(LoadPredictionEvent event) {
                        return event.isLargeError();
                    }
                })
                .within(Time.minutes(30)); // 30分钟内的连续5次误差
        
        // 应用模式到数据流
        PatternStream<LoadPredictionEvent> patternStream = CEP.pattern(eventStream, largeErrorPattern);
        
        // 处理匹配的模式
        return patternStream.select(new PatternSelectFunction<LoadPredictionEvent, String>() {
            @Override
            public String select(Map<String, List<LoadPredictionEvent>> pattern) throws Exception {
                List<LoadPredictionEvent> firstEvent = pattern.get("first");
                List<LoadPredictionEvent> fifthEvent = pattern.get("fifth");
                
                if (firstEvent != null && !firstEvent.isEmpty() && 
                    fifthEvent != null && !fifthEvent.isEmpty()) {
                    
                    LoadPredictionEvent first = firstEvent.get(0);
                    LoadPredictionEvent fifth = fifthEvent.get(0);
                    
                    return String.format("模型预测告警 - 连续5次误差较大！时间范围: %s 到 %s, 最新误差: %.2f%%, 真实负荷: %.2f, 预测负荷: %.2f",
                            first.getTrueData().getTimestamp(),
                            fifth.getTrueData().getTimestamp(),
                            fifth.getRelativeErrorPercentage(),
                            fifth.getTrueData().getLoad(),
                            fifth.getPredictData().getPredictedLoad());
                }
                
                return "模型预测告警 - 连续5次误差较大！";
            }
        });
    }
    
    /**
     * 计算滑动窗口内的MAE和RMSE
     */
    public static DataStream<String> calculateMetrics(DataStream<LoadPredictionEvent> eventStream) {
        return eventStream
                .keyBy(event -> "metrics") // 所有事件使用同一个key
                .countWindow(10, 1) // 滑动窗口：每10个事件计算一次，每次滑动1个事件
                .apply(new org.apache.flink.streaming.api.functions.windowing.WindowFunction<LoadPredictionEvent, String, String, org.apache.flink.streaming.api.windowing.windows.GlobalWindow>() {
                    @Override
                    public void apply(String key, 
                                    org.apache.flink.streaming.api.windowing.windows.GlobalWindow window, 
                                    Iterable<LoadPredictionEvent> events, 
                                    org.apache.flink.util.Collector<String> out) throws Exception {
                        double totalAbsoluteError = 0.0;
                        double totalSquaredError = 0.0;
                        int count = 0;
                        
                        LoadPredictionEvent latestEvent = null;
                        
                        for (LoadPredictionEvent event : events) {
                            totalAbsoluteError += event.getAbsoluteError();
                            totalSquaredError += event.getSquaredError();
                            count++;
                            latestEvent = event;
                        }
                        
                        if (count > 0 && latestEvent != null) {
                            double mae = totalAbsoluteError / count;
                            double rmse = Math.sqrt(totalSquaredError / count);
                            
                            // 检查最新事件是否为大误差
                            if (!latestEvent.isLargeError()) {
                                String result = String.format("当前MAE: %.2f, RMSE: %.2f (基于最近%d个预测) - 时间: %s, 真实负荷: %.2f, 预测负荷: %.2f, 误差: %.2f%%",
                                        mae, rmse, count,
                                        latestEvent.getTrueData().getTimestamp(),
                                        latestEvent.getTrueData().getLoad(),
                                        latestEvent.getPredictData().getPredictedLoad(),
                                        latestEvent.getRelativeErrorPercentage());
                                out.collect(result);
                            }
                        }
                    }
                });
    }
}